#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <string.h>
#include <unistd.h>
#include "common.h"
#include "schema.h"
#include "gpulib.h"
#include "cpulib.h"

#define OFFSETOF(type, field)    ((unsigned long) &(((type *) 0)->field))
#define BLOCK	(1024*1024*50)

extern void tableScan(struct tableNode *, struct relation *, struct statistic *);
extern struct tableNode * hashJoin(struct joinNode *, struct statistic *);

int main(int argc, char ** argv){

	int * tmp;
	cudaMalloc((void **)&tmp, 4);
	cudaFree(tmp);

	struct timespec start,end;
	struct timespec startAll;
	struct statistic pp;

	clock_gettime(CLOCK_REALTIME,&startAll);

	double totalTime = 0;


	struct tableNode *dateTable = NULL;
	struct tableNode *loTable = NULL;
	struct tableNode * partTable = NULL;
	struct tableNode * supplierTable = NULL;
	struct tableNode * customerTable = NULL;
	int outFd;
	long outSize;
	char *outTable;

	dateTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	dateTable->totalAttr = 2;
	dateTable->attrType = (int *) malloc(sizeof(int)*dateTable->totalAttr);
	dateTable->attrType[0] = INT;
	dateTable->attrType[1] = INT;
	dateTable->attrSize = (int *) malloc(sizeof(int)*dateTable->totalAttr);
	dateTable->attrSize[0] = sizeof(int);
	dateTable->attrSize[1] = sizeof(int);
	dateTable->tupleSize = 2*sizeof(int);
	dateTable->content = (char **) malloc(sizeof(char *) * dateTable->totalAttr);
	outFd = open("DDATE4",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	dateTable->tupleNum = outSize/sizeof(int);
	dateTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(dateTable->content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);

	outFd = open("DDATE0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	dateTable->content[1] = (char *) malloc(outSize);
	memcpy(dateTable->content[1],outTable,outSize);	
	munmap(outTable,outSize);
	close(outFd);
	dateTable->dataPos = MEM;

	partTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	partTable->totalAttr = 1;
	partTable->attrType = (int *) malloc(sizeof(int) * partTable->totalAttr);
	partTable->attrType[0] = INT;
	partTable->attrSize = (int *) malloc(sizeof(int) * partTable->totalAttr);
	partTable->attrSize[0] = sizeof(int);
	partTable->tupleSize = sizeof(int);
	partTable->content = (char **) malloc(sizeof(char *) *partTable->totalAttr);
	outFd = open("PART0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	partTable->content[0] = (char *) malloc(outSize);
	memcpy(partTable->content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);
	partTable->tupleNum = outSize / sizeof(int);

	struct relation partRel;
	partRel.hasWhere = 1;
	partRel.whereAttrNum = 1;
	partRel.whereAttrType = (int *)malloc(sizeof(int) * partRel.whereAttrNum);
	partRel.whereAttrType[0] = STRING;
	partRel.whereAttrSize = (int*) malloc(sizeof(int) * partRel.whereAttrNum);
	partRel.whereAttrSize[0] = 6;
	outFd = open("PART2",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	partRel.content = (char **) malloc(sizeof(char *) * partRel.whereAttrNum);
	partRel.content[0] = (char *) malloc(outSize);
	memcpy(partRel.content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);

	partRel.filter = (struct whereCondition *) malloc(sizeof(struct whereCondition));
	(partRel.filter)->nested = 0;
	(partRel.filter)->andOr = OR;
	(partRel.filter)->exp[0].index = 0;
	(partRel.filter)->exp[0].relation = EQUAL;
	strcpy((partRel.filter)->exp[0].content,"MFGR#1");
	(partRel.filter)->exp[1].index = 0;
	(partRel.filter)->exp[1].relation = EQUAL;
	strcpy((partRel.filter)->exp[1].content,"MFGR#2");

	clock_gettime(CLOCK_REALTIME,&start);
	tableScan(partTable, &partRel, &pp);
	clock_gettime(CLOCK_REALTIME,&end);

        totalTime += (end.tv_sec -  start.tv_sec)* BILLION + end.tv_nsec - start.tv_nsec;

	supplierTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	supplierTable->totalAttr = 1;
	supplierTable->attrType = (int *) malloc(sizeof(int)* supplierTable->totalAttr);
	supplierTable->attrType[0] = INT;
	supplierTable->attrSize = (int *) malloc(sizeof(int)* supplierTable->totalAttr);
	supplierTable->attrSize[0] = sizeof(int);
	supplierTable->tupleSize = sizeof(int);
	outFd = open("SUPPLIER0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	supplierTable->content = (char **) malloc(sizeof(char *) * supplierTable->totalAttr);
	supplierTable->content[0] = (char *) malloc(outSize);
	memcpy(supplierTable->content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);
	supplierTable->tupleNum = outSize / sizeof(int) ;

	struct relation supRel;

	supRel.hasWhere = 1;
	supRel.whereAttrNum = 1;
	supRel.whereAttrType = (int *) malloc(sizeof(int) * supRel.whereAttrNum);
	supRel.whereAttrType[0] = STRING;
	supRel.whereAttrSize = (int *) malloc(sizeof(int) * supRel.whereAttrNum);
	supRel.whereAttrSize[0] = 12;
	outFd = open("SUPPLIER5",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	supRel.content = (char **) malloc(sizeof(char *) * supRel.whereAttrNum);
	supRel.content[0] = (char *) malloc(outSize);
	memcpy(supRel.content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);

	supRel.filter = (struct whereCondition *) malloc(sizeof(struct whereCondition));
	(supRel.filter)->nested = 0;
	(supRel.filter)->andOr = EXP;
	(supRel.filter)->exp[0].index = 0;
	(supRel.filter)->exp[0].relation = EQUAL;
	strcpy((supRel.filter)->exp[0].content,"AMERICA");

	clock_gettime(CLOCK_REALTIME,&start);
	tableScan(supplierTable, &supRel, &pp);
	clock_gettime(CLOCK_REALTIME,&end);

        totalTime += (end.tv_sec -  start.tv_sec)* BILLION + end.tv_nsec - start.tv_nsec;

	customerTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	customerTable->totalAttr = 2;
	customerTable->attrType = (int *) malloc(sizeof(int) * customerTable->totalAttr);
	customerTable->attrSize = (int *) malloc(sizeof(int) * customerTable->totalAttr);
	customerTable->attrType[0] = INT;
	customerTable->attrType[1] = STRING;
	customerTable->attrSize[0] = sizeof(int);
	customerTable->attrSize[1] = 15;
	customerTable->tupleSize = 19;
	customerTable->content = (char **) malloc(sizeof(char *) * customerTable->totalAttr);
	outFd = open("CUSTOMER0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	customerTable->content[0] = (char *) malloc(outSize);
	memcpy(customerTable->content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);
	customerTable->tupleNum = outSize / sizeof(int);

	outFd = open("CUSTOMER4",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	customerTable->content[1] = (char *) malloc(outSize);
	memcpy(customerTable->content[1],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);

	struct relation cusRel;

	cusRel.hasWhere = 1;
	cusRel.whereAttrNum = 1;
	cusRel.whereAttrType = (int *) malloc(sizeof(int) * cusRel.whereAttrNum);
	cusRel.whereAttrType[0] = STRING;
	cusRel.whereAttrSize = (int *) malloc(sizeof(int) * cusRel.whereAttrNum);
	cusRel.whereAttrSize[0] = 12;
	outFd = open("CUSTOMER5",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	cusRel.content = (char **) malloc(cusRel.whereAttrNum * sizeof(char *));
	cusRel.content[0] = (char *) malloc(outSize);
	memcpy(cusRel.content[0],outTable,outSize);
	munmap(outTable,outSize);
	close(outFd);

	cusRel.filter = (struct whereCondition *) malloc(sizeof(struct whereCondition));
	(cusRel.filter)->nested = 0;
	(cusRel.filter)->andOr = EXP;
	(cusRel.filter)->exp[0].index = 0;
	(cusRel.filter)->exp[0].relation = EQUAL;
	strcpy((cusRel.filter)->exp[0].content,"AMERICA");

	clock_gettime(CLOCK_REALTIME,&start);
	tableScan (customerTable, &cusRel, &pp);
	clock_gettime(CLOCK_REALTIME,&end);

        totalTime += (end.tv_sec -  start.tv_sec)* BILLION + end.tv_nsec - start.tv_nsec;

	loTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	loTable->totalAttr = 6;
	loTable->attrType = (int *) malloc(sizeof(int) * loTable->totalAttr);
	loTable->attrSize = (int *) malloc(sizeof(int) * loTable->totalAttr);
	loTable->content = (char **) malloc(sizeof(char *) * loTable->totalAttr);
	loTable->attrType[0] = INT;
	loTable->attrType[1] = INT;
	loTable->attrType[2] = INT;
	loTable->attrType[3] = INT;
	loTable->attrType[4] = INT;
	loTable->attrType[5] = INT;
	loTable->attrSize[0] = sizeof(int);
	loTable->attrSize[1] = sizeof(int);
	loTable->attrSize[2] = sizeof(int);
	loTable->attrSize[3] = sizeof(int);
	loTable->attrSize[4] = sizeof(int);
	loTable->attrSize[5] = sizeof(int);
	loTable->tupleSize = 6 * sizeof(int);

	outFd = open("LINEORDER12",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	loTable->tupleNum = outSize / sizeof(int);
	close(outFd);

	struct joinNode jNode;

	jNode.dimNum = 4;

	jNode.factTable = loTable;
	jNode.dimTable = (struct tableNode **) malloc(sizeof(struct tableNode *) * jNode.dimNum); 
	jNode.factIndex = (int *) malloc(sizeof(int)*jNode.dimNum);
	jNode.dimIndex = (int *) malloc(sizeof(int)*jNode.dimNum);

	jNode.dimTable[0] = supplierTable;
	jNode.dimIndex[0] = 0;
	jNode.factIndex[0] = 3;

	jNode.dimTable[1] = customerTable;
	jNode.dimIndex[1] = 0;
	jNode.factIndex[1] = 2; 

	jNode.dimTable[2] = partTable;
	jNode.dimIndex[2] = 0;
	jNode.factIndex[2] = 4;

	jNode.dimTable[3] = dateTable;
	jNode.dimIndex[3] = 1;
	jNode.factIndex[3] = 5;

	jNode.totalAttr = 4;
	jNode.tupleSize = 27;
	jNode.attrType = (int *) (malloc(sizeof(int) * jNode.totalAttr));
	jNode.attrSize = (int *) (malloc(sizeof(int) * jNode.totalAttr));
	jNode.attrType[0] = INT;
	jNode.attrType[1] = STRING;
	jNode.attrType[2] = INT;
	jNode.attrType[3] = INT;
	jNode.attrSize[0] = sizeof(int);
	jNode.attrSize[1] = 15;
	jNode.attrSize[2] = sizeof(int);
	jNode.attrSize[3] = sizeof(int);

	jNode.factOutNum = 2;
	jNode.factOutIndex = (int *) malloc(jNode.factOutNum * sizeof(int));
	jNode.factOutIndex[0] = 0;
	jNode.factOutIndex[1] = 1;

	jNode.dimOutTotal = 2;
	jNode.dimOutNum = (int *) malloc(jNode.dimNum * sizeof(int));
	jNode.dimOutNum[0] = 0;
	jNode.dimOutNum[1] = 1;
	jNode.dimOutNum[2] = 0;
	jNode.dimOutNum[3] = 1;

	jNode.dimOutIndex = (int **) malloc(jNode.dimNum * sizeof(int *));
	jNode.dimOutIndex[1] = (int *) malloc(jNode.dimOutNum[1] * sizeof(int));
	jNode.dimOutIndex[1][0] = 1;
	jNode.dimOutIndex[3] = (int *) malloc(jNode.dimOutNum[3] * sizeof(int));
	jNode.dimOutIndex[3][0] = 0;

	int pass = loTable->tupleNum / BLOCK + 1;
	long tupleUnit = loTable->tupleNum / pass;
	long nextScan = tupleUnit;
	long restTuple = loTable->tupleNum;
	long tupleOffset = 0;
	long offset = 0; 

	for(int i=0;i<pass;i++){
		if(restTuple < nextScan)
			nextScan = restTuple;

		loTable->tupleNum = nextScan;

		outFd = open("LINEORDER12",O_RDONLY);
		offset = tupleOffset * sizeof(int);
		outSize = nextScan * sizeof(int);
		outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);
		loTable->content[0] = (char *) malloc(outSize);
		memcpy(loTable->content[0],outTable + offset,outSize);
		munmap(outTable,outSize);
		close(outFd);

		outFd = open("LINEORDER13",O_RDONLY);
		offset = tupleOffset * sizeof(int);
		outSize = nextScan * sizeof(int);
		outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);
		loTable->content[1] = (char *) malloc(outSize);
		memcpy(loTable->content[1],outTable + offset,outSize);
		munmap(outTable,outSize);
		close(outFd);

		outFd = open("LINEORDER2",O_RDONLY);
		offset = tupleOffset * sizeof(int);
		outSize = nextScan * sizeof(int);
		outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);
		loTable->content[2] = (char *) malloc(outSize);
		memcpy(loTable->content[2],outTable + offset,outSize);
		munmap(outTable,outSize);
		close(outFd);

		outFd = open("LINEORDER4",O_RDONLY);
		offset = tupleOffset * sizeof(int);
		outSize = nextScan * sizeof(int);
		outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);
		loTable->content[3] = (char *) malloc(outSize);
		memcpy(loTable->content[3],outTable + offset ,outSize);
		munmap(outTable,outSize);
		close(outFd);

		outFd = open("LINEORDER3",O_RDONLY);
		offset = tupleOffset * sizeof(int);
		outSize = nextScan * sizeof(int);
		outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);
		loTable->content[4] = (char *) malloc(outSize);
		memcpy(loTable->content[4],outTable + offset,outSize);
		munmap(outTable,outSize);
		close(outFd);

		outFd = open("LINEORDER5",O_RDONLY);
		offset = tupleOffset * sizeof(int);
		outSize = nextScan * sizeof(int);
		outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);
		loTable->content[5] = (char *) malloc(outSize);
		memcpy(loTable->content[5],outTable + offset,outSize);
		munmap(outTable,outSize);
		close(outFd);

		clock_gettime(CLOCK_REALTIME,&start);

		struct tableNode *join1 = hashJoin(&jNode, & pp);

		free(join1);

		free(loTable->content[0]);
		free(loTable->content[1]);
		free(loTable->content[2]);
		free(loTable->content[3]);
		free(loTable->content[4]);
		free(loTable->content[5]);

		tupleOffset += tupleUnit;
		restTuple -= nextScan;

		clock_gettime(CLOCK_REALTIME,&end);

        	totalTime += (end.tv_sec -  start.tv_sec)* BILLION + end.tv_nsec - start.tv_nsec;
	}

	clock_gettime(CLOCK_REALTIME,&end);
        double timeE = (end.tv_sec -  startAll.tv_sec)* BILLION + end.tv_nsec - startAll.tv_nsec;
        printf("Total query CPU time: %lf\n", timeE);	
	printf("Total time without Disk IO: %lf\n", totalTime);
	printf("Dim Transfer %lf\n",pp.dimTransfer);
	printf("Fact Transfer %lf\n",pp.factTransfer);
	printf("Total Transfer %lf\n", pp.totalTransfer);
	
	return 0;
}
